package beanstalk

import (
	"fmt"
	"log"
	"sync"
	"time"

	"xp/lib/beanstalk/gobeanstalk"
	"xp/lib/finalize"
	"xp/lib/utils"
)

const (
	pri        int = 1024
	defaultTTR int = 60
)

type Job struct {
	c    *Consumer
	conn *conn
	uri  string
	*gobeanstalk.Job
}

func (job *Job) Delete() error {
	if err := job.c.DeleteJob(job.Id, job.uri); err != nil {
		log.Println("delete job error:", err)
	}
	return nil
}

type put struct {
	tube  string
	delay int
	ttr   int
	body  []byte
	c     *Producer
}

type conn struct {
	sync.Mutex
	*gobeanstalk.Conn

	uri string
	// tube current using
	using    string
	client   *client
	reserved chan *Job
	magic    int64
	putChan  chan *put
	closed   bool
}

func newConn(uri string, client *client) (*conn, error) {
	c, err := gobeanstalk.Dial(uri)
	if err != nil {
		return nil, err
	}
	conn := &conn{
		uri:     uri,
		Conn:    c,
		client:  client,
		magic:   utils.NewGUID(),
		putChan: make(chan *put, 1024),
		closed:  false,
	}
	return conn, nil
}

func (c *conn) Close() error {
	if c.client != nil {
		if c.magic != 0 {
			c.client.connCloseChan <- c.magic
		}

	}
	c.close()
	return nil
}

func (c *conn) putJob(put *put) bool {
	c.Lock()
	defer c.Unlock()
	if !c.closed {
		c.putChan <- put
	}
	return !c.closed
}

func (c *conn) close() {
	c.Lock()
	defer c.Unlock()
	if c.closed {
		return
	}
	c.closed = true
	for len(c.putChan) != 0 {
		time.Sleep(time.Millisecond)
	}
	close(c.putChan)
	c.Conn.Close()
}

func (c *conn) put(tube string, ttr, delay int, body []byte) error {
	if c.reserved != nil {
		return fmt.Errorf("can not use Consumer as producer")
	}
	if tube != c.using {
		if err := c.Use(tube); err != nil {
			return err
		}
		c.using = tube
	}
	if _, err := c.Put(body, pri, delay, ttr); err != nil {
		return err
	}
	return nil
}

func (c *conn) returnToClient(put *put) {
	c.Close()
	if put != nil {
		put.c.PutTube(put.tube, put.body)
	} else if c.client != nil {
		c.client.newChan <- c.uri
	}
}

func (c *conn) putLoop() {
	for put := range c.putChan {
		if err := c.put(put.tube, put.ttr, put.delay, put.body); err != nil {
			fmt.Println("beanstalk put error:", err)
			c.returnToClient(put)
		}
	}
}

func (c *conn) release(id uint64, delay int) error {
	c.Lock()
	defer c.Unlock()
	if c.closed {
		return nil
	}
	return c.Release(id, pri, delay)
}

func (c *conn) reserve(consumer *Consumer) {
	for {
		job, err := c.Reserve()
		if err != nil {
			c.returnToClient(nil)
			log.Println("reserve error:", err)
			return
		}
		if job == nil {
			continue
		}
		if err := c.release(job.Id, defaultTTR); err != nil {
			c.returnToClient(nil)
			log.Println("reserve error:", err)
			return
		}
		c.reserved <- &Job{c: consumer, uri: c.uri, Job: job}
	}
}

func (c *conn) watch(tube string) error {
	c.Lock()
	defer c.Unlock()
	c.using = tube
	if _, err := c.Watch(tube); err != nil {
		return err
	}
	return nil
}

type client struct {
	conns         chan *conn
	newChan       chan string
	quitChan      chan struct{}
	loopChan      chan struct{}
	delay         int
	ttr           int
	holding       map[int64]*conn
	connCloseChan chan int64
	producer      bool
}

func newClient(sockets []string, maxConn int, tmp, producer bool) *client {
	c := &client{
		producer:      producer,
		holding:       make(map[int64]*conn),
		quitChan:      make(chan struct{}, 1),
		loopChan:      make(chan struct{}, 1),
		ttr:           defaultTTR,
		conns:         make(chan *conn, maxConn*len(sockets)),
		newChan:       make(chan string, maxConn*len(sockets)),
		connCloseChan: make(chan int64, maxConn*len(sockets)),
	}
	go c.loop()
	for i := 0; i < maxConn*len(sockets); i++ {
		c.newChan <- sockets[i%len(sockets)]
	}
	if !tmp {
		finalize.OnExit(c.Close)
	}
	return c
}

func (c *client) Close() error {
	c.quitChan <- struct{}{}
	<-c.loopChan
	for _, conn := range c.holding {
		conn.close()
	}
	return nil
}

func (c *client) loop() {
	defer func() {
		c.loopChan <- struct{}{}
	}()
	for {
		select {
		case target := <-c.newChan:
			for {
				conn, err := newConn(target, c)
				if err != nil {
					fmt.Println("trying connect error:", err, target)
					time.Sleep(5 * time.Second)
					continue
				}
				select {
				case c.conns <- conn:
					c.holding[conn.magic] = conn
					if c.producer {
						go conn.putLoop()
					}
				default:
					conn.magic = 0
					conn.close()
				}
				break
			}
		case <-c.quitChan:
			fmt.Println("closing beanstalk client...")
			return
		case magic := <-c.connCloseChan:
			if _, ok := c.holding[magic]; ok {
				delete(c.holding, magic)
			}
		}
	}
}

func (c *client) setParams(delay int) {
	c.delay = delay
}

type Producer struct {
	*client

	tube string
}

func (p *Producer) SetTTR(val int) {
	p.client.ttr = val
}

func NewProducerWithDelay(sockets []string, tube string, delay int) *Producer {
	p := &Producer{
		tube:   tube,
		client: newClient(sockets, 1, false, true),
	}
	p.client.setParams(delay)
	return p
}

func NewProducer(sockets []string, tube string) *Producer {
	return NewProducerWithDelay(sockets, tube, 0)
}

func (c *Producer) Put(body []byte) error {
	return c.PutTube(c.tube, body)
}

// Write implements io.Writer
func (c *Producer) Write(body []byte) (int, error) {
	if err := c.Put(body); err != nil {
		return 0, err
	}
	return len(body), nil
}

func (c *Producer) PutTube(tube string, body []byte) error {
	if tube == "" {
		return nil
	}
	for {
		select {
		case conn := <-c.conns:
			put := &put{tube, c.delay, c.client.ttr, body, c}
			if conn.putJob(put) {
				c.conns <- conn
				return nil
			} else {
				c.newChan <- conn.uri
			}
		}
	}
}

func (c *Producer) PutTubeBytes(tube string, body []byte) error {
	return c.PutTube(tube, body)
}

type Consumer struct {
	sync.Mutex
	*client

	tube     string
	reserved chan *Job
	cmd      map[string]*gobeanstalk.Conn
}

func NewConsumerSize(sockets []string, tube string, size int) *Consumer {
	c := &Consumer{
		tube:     tube,
		cmd:      make(map[string]*gobeanstalk.Conn),
		client:   newClient(sockets, 1, false, false),
		reserved: make(chan *Job, size),
	}
	go c.loop()
	return c
}

func NewConsumer(sockets []string, tube string) *Consumer {
	return NewConsumerSize(sockets, tube, 2)
}

func NewTmpConsumer(sockets []string, tube string, reserve bool) *Consumer {
	c := &Consumer{
		tube:     tube,
		cmd:      make(map[string]*gobeanstalk.Conn),
		client:   newClient(sockets, 1, true, false),
		reserved: make(chan *Job, 1),
	}
	c.client.ttr = 1
	if reserve {
		go c.loop()
	}
	return c
}

func (c *Consumer) loop() {
	for conn := range c.conns {
		if err := conn.watch(c.tube); err != nil {
			fmt.Println("watching tube failed:", err)
			time.Sleep(time.Second)
			c.newChan <- conn.uri
			continue
		}
		conn.reserved = c.reserved
		go conn.reserve(c)
	}
}

func (c *Consumer) getCmdConn(uri string) (*gobeanstalk.Conn, error) {
	var err error
	conn, ok := c.cmd[uri]
	if !ok {
		conn, err = gobeanstalk.Dial(uri)
		if err != nil {
			return nil, err
		}
	}
	return conn, nil
}

func (c *Consumer) DeleteJob(id uint64, uri string) error {
	c.Lock()
	defer c.Unlock()

	conn, err := c.getCmdConn(uri)
	if err != nil {
		return err
	}

	if err := conn.Delete(id); err != nil {
		conn.Close()
		delete(c.cmd, uri)
		return err
	}
	c.cmd[uri] = conn
	return nil
}

func (c *Consumer) Reserve() <-chan *Job {
	return c.reserved
}
